package com.lion.queue.runner;

import java.util.Iterator;

import com.lion.queue.config.Configuration;
import com.lion.tools.GenericsTools;

/**
 * 队列消费者
 * 
 * @note 当队列中没有元素时，等待
 * @author liyongyao
 *
 * @param <T>
 */
public abstract class QueueRunner<T,R> implements Runnable {

	protected Class<T> clazz;
	protected Queue<T> queue;
	protected Status status;
	protected Long waitingTime = 0l;
	protected String queueName;
	protected RunnerCallBack<T,R> callBack;
	private final int NOMAL_PRIORITY = 3;

	@SuppressWarnings("unchecked")
	public Class<T> instanceAnnotation() {
		try {
			return GenericsTools.getSuperClassGenricType(getClass(), 0);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}

	public QueueRunner(String queueName) {
		this.clazz = instanceAnnotation();
		this.queueName = queueName;
		this.status = Status.RUNNING;
		int maxSize = Configuration.getInstance().getInt("queue.max.size", 100);
		this.queue = QueueManager.getQueueManager().createQueue(queueName, maxSize, clazz);
	}

	/**
	 * 指定优先级添加任务，越大优先级越高
	 * 
	 * @param entity
	 * @param priority
	 */
	public void add(T entity, int priority) {
		queue.add(entity, priority);
	}

	/**
	 * 添加任务，默认优先级3
	 * 
	 * @param entity
	 */
	public void add(T entity) {
		queue.add(entity, NOMAL_PRIORITY);
	}

	/**
	 * 取得当前最大优先级
	 * 
	 * @return
	 */
	public Integer getMaxPriority() {
		Integer maxPriority = NOMAL_PRIORITY;
		if (queue != null) {
			Iterator<QueueItem<T>> ite = queue.getBlockingQueue().iterator();
			while (ite.hasNext()) {
				QueueItem<T> item = ite.next();
				if (item.getPriority() > maxPriority.intValue()) {
					maxPriority = item.getPriority();
				}
			}
		}
		return maxPriority;
	}

	public enum Status {
		RUNNING, WAITING, STOP
	}

	public boolean isRunning() {
		return this.status == Status.RUNNING;
	}

	public boolean isWaiting() {
		return this.status == Status.WAITING;
	}

	public boolean isStop() {
		return this.status == Status.STOP;
	}

	/**
	 * 具体执行业务
	 * 
	 * @param task
	 */
	protected abstract R execute(T task);

	@Override
	public void run() {
			Long maxWarttime = Configuration.getInstance().getLong("queue.max.warttime", 30 * 1000L);
			Long sleepTime = Configuration.getInstance().getLong("queue.fetch.interval",1000L);
			boolean enableExit = Configuration.getInstance().getBoolean("thread.enable.exit",true);
			while (true) {
				if (waitingTime > maxWarttime && enableExit) {// 超过等待时间，退出线程
					System.out.println("[" + queueName + "]队列等待时间超过[" + maxWarttime + "]毫秒,退出线程");
					this.status = Status.STOP;
					waitingTime = 0l;
					break;
				} else {
					QueueItem<T> item = queue.poll();
					if (item == null) {
						try {
							Thread.sleep(sleepTime);
						} catch (Exception e) {
						}
						waitingTime += sleepTime;
					} else {
						this.status = Status.RUNNING;
						R result = null;
						Exception e = null;
						try {
							result = execute(item.getEntity());
						} catch (Exception e1) {
							e = e1;
						}finally{
							if(this.callBack != null){
								RunnerResult<R> r = new RunnerResult<R>();
								r.setResult(result);
								r.setE(e);
								if(e != null){
									r.setError(true);
								}
								callBack.callBack(item.getEntity(),r);
							}
						}
						this.status = Status.WAITING;
						waitingTime = 0l;
					}
				}
			}
	}

	public Queue<T> getQueue() {
		return queue;
	}

	public void setQueue(Queue<T> queue) {
		this.queue = queue;
	}

	public Status getStatus() {
		return status;
	}

	public void setStatus(Status status) {
		this.status = status;
	}

	public Long getWaitingTime() {
		return waitingTime;
	}

	public void setWaitingTime(Long waitingTime) {
		this.waitingTime = waitingTime;
	}

	public void setCallBack(RunnerCallBack<T,R> callBack) {
		this.callBack = callBack;
	}
	
	
}
